package com.lin.flow;


import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.concurrent.*;

/**
 * @author: linjiang
 * @date: 2025/5/4
 */
public class JDK9FlowDemo {

    /**
     * 测试流处理器
     *
     * @throws IOException
     */
    @Test
    public void testProcessor() throws IOException {
        SubmissionPublisher publisher = new SubmissionPublisher<>();

        // 订阅者
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription; // processer 和 subscriber 之间的订阅关联

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                if (item % 8 == 0) {
                    System.out.println("subscriber 收到" + item + "数据。");
                }

                if (item > 50) {
                    int i = 1 / 0;
                }
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("subscriber 出现异常 ：" + throwable.getMessage());
            }

            @Override
            public void onComplete() {

            }
        };

        // 第一个processor
        MyProcesser processor = new MyProcesser();

        publisher.subscribe(processor);
        processor.subscribe(subscriber);

        for (int i = 0; i < 100; i++) {
            publisher.submit(i);
        }

        System.in.read();
    }

    /**
     * 自定义流处理器
     */
    public class MyProcesser extends SubmissionPublisher<Integer> implements Flow.Processor<Integer, Integer> {
        Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(Integer item) {
            int i = item * 2;
            System.out.println("processor 收到" + i + "数据。");
            this.submit(item);
            subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {

        }

    }


    /**
     * 测试异常回调方法或流正常关闭时触发的方法。
     *
     * @throws IOException
     */
    @Test
    public void testBackCall() throws IOException {
        SubmissionPublisher<Object> publisher = new SubmissionPublisher<>();

        // 订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber<Integer>() {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                System.out.println(Thread.currentThread().getName() + "收到数据：" + item);
//                if (item == 500) {
//                    int i = 1 / 0;
//                }
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("在处理流数据的过程中，出现了异常：" + throwable.getMessage());
            }

            /**
             * 发布者关闭流之后，触发回调方法
             */
            @Override
            public void onComplete() {
                System.out.println("生产者正常关闭。。。。。");
            }
        };


        publisher.subscribe(subscriber);
        for (int i = 0; i < 1000; i++) {
            publisher.submit(i);
        }

        // 发布者关闭
        publisher.close();
    }


    /**
     * 测试背压模式
     *
     * @throws IOException
     */
    @Test
    public void testBackPress() {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        // 设置缓存区的大小，默认大小256
        SubmissionPublisher<Object> publisher = new SubmissionPublisher<>(executorService, Flow.defaultBufferSize());

        // 订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1); // 一次获取一个数据
            }

            @Override
            public void onNext(Object item) {
                System.out.println(Thread.currentThread().getName() + "收到数据：" + item);

                // 模拟订阅者，消费能力不足。
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    System.out.println("睡眠异常。");
                }
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        };


        // 绑定
        publisher.subscribe(subscriber);

        // 生产数据
        for (int i = 0; i < 1000; i++) {
            /*
             * 效果：
             * 会在256个元素的时候，缓存池的大小被占满。
             * 之后订阅来不及消费时，就会向生产者发送消息。订阅者收到消息之后，就会放慢生产数据的速率。
             * 防止生产过快，订阅者的数据消费能力不够，造成内存溢出。
             * */
            System.out.println(Thread.currentThread().getName() + "生产数据：" + i);
            publisher.submit(i);
            System.out.println(Thread.currentThread().getName() + "生产数据完毕：" + i);
        }
        // 关闭线程
        executorService.shutdown();
    }


    @Test
    public void testBasic() throws IOException {
        // 定义响应式发布者
        SubmissionPublisher publisher = new SubmissionPublisher();
        // 响应式消费者
        Flow.Subscriber<Object> subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription;

            /**
             * 消费者订阅时，触发
             * @param subscription 一次新的订阅
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + "订阅");
                this.subscription = subscription;
                subscription.request(1); // 向生成者发送请求N个数据的消息
            }

            @Override
            public void onNext(Object item) {
                System.out.println(Thread.currentThread().getName() + "收到数据：" + item);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("在处理流数据的过程中，出现了异常：" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("消费者处理完所有的流数据。");
            }
        };

        // 消费者订阅
        publisher.subscribe(subscriber);


        // 生成者生产数据。
        for (int i = 0; i < 10; i++) {
            System.out.println(Thread.currentThread().getName() + "生产数据：" + i);
            publisher.submit(i);
            System.out.println(Thread.currentThread().getName() + "生产数据完毕：" + i);
        }

        System.in.read();
    }

}
